package com.example.kafka.demo;

import com.example.kafka.constant.KafkaConstant;
import com.example.kafka.service.Producer;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Random;

/**
 * @application:
 * @author: LiYuhang
 * @Date: 2021/6/26 15:06
 * @version: 0.1
 * @Description :
 * @Modified By:
 */
@Slf4j
public class ProducerDemo1 implements Producer {
    private final KafkaProducer<String, String> producer;
    private final String topic;

    public ProducerDemo1(String topic) {
        this.producer = new KafkaProducer<String, String>(Producer.getDefaultConfig());
        this.topic = topic;
    }

    @Override
    public void process(String message) {
        try {
            RecordMetadata metadata = Producer.write(producer,this.topic,message);
            log.info("offset:{},partition:{},partition:{}",metadata.offset(),metadata.partition(),metadata.timestamp());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void  run(){
        while (true){
            int nextInt = new Random().nextInt(10);
            for (int i = 0; i < nextInt; i++) {
                process(KafkaConstant.generatorUuid());
            }
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
            }
        }
    }


    public static void main(String[] args) {
        ProducerDemo1 demo1=new ProducerDemo1("test");
        demo1.run();
    }
}
